package org.spark.api.util.hbase;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.spark.api.config.HBaseInstance;
import org.spark.api.util.ReflectHBaseUtils;
import org.spark.api.util.model.HBaseClassMapping;
import org.spark.api.util.model.HBaseColumnMapping;
import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @param <T>
 * @author chenwy
 */
public abstract class AbstractHBaseDao<T> implements HBaseDao<T> {

    private HBaseInstance instance;

    private Class<T> clazz;

    public AbstractHBaseDao(HBaseInstance instance) {
        this.instance = instance;
        //获取泛型类的class
        Class<T> clazz = (Class<T>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
        this.clazz = clazz;
    }

    protected Get buildGet(byte[] rowKey, Map<String, HBaseColumnMapping> columnMapping) {
        Get get = new Get(rowKey);
        if (columnMapping != null) {
            for (Map.Entry<String, HBaseColumnMapping> mapping : columnMapping.entrySet()) {
                HBaseColumnMapping column = mapping.getValue();
                get.addColumn(column.getFamily(), column.getQualifier());
            }
        }
        return get;
    }

    protected Scan buildScan(Map<String, HBaseColumnMapping> columnMapping) {
        Scan scan = new Scan();
        scan.setCacheBlocks(false);
        scan.setCaching(1000);
        if (columnMapping != null) {
            for (Map.Entry<String, HBaseColumnMapping> mapping : columnMapping.entrySet()) {
                HBaseColumnMapping column = mapping.getValue();
                scan.addColumn(column.getFamily(), column.getQualifier());
            }
        }
        return scan;
    }

    protected Scan buildScan(Map<String, HBaseColumnMapping> columnMapping, byte[] startRow, byte[] stopRow) {
        Scan scan = buildScan(columnMapping)
                .withStartRow(startRow)
                .withStopRow(stopRow);
        return scan;
    }

    protected Scan buildScan(Map<String, HBaseColumnMapping> columnMapping, byte[] rowPrefix) {
        Scan scan = buildScan(columnMapping)
                .setRowPrefixFilter(rowPrefix);
        return scan;
    }

    @Override
    public T get(TableName tableName, byte[] rowKey) {

        HBaseClassMapping classMapping = ReflectHBaseUtils.getHBaseClassMapping(clazz);
        try (
                Connection connection = instance.getConnection();
                Table table = connection.getTable(tableName)
        ) {
            Get get = buildGet(rowKey, classMapping.getAll());
            Result result = table.get(get);
            return ReflectHBaseUtils.parseObject(result, clazz);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public List<T> get(TableName tableName, List<byte[]> rowKeys) {
        HBaseClassMapping classMapping = ReflectHBaseUtils.getHBaseClassMapping(clazz);
        List<T> list = new ArrayList<>();
        try (
                Connection connection = instance.getConnection();
                Table table = connection.getTable(tableName)
        ) {
            for (byte[] rowKey : rowKeys) {
                Get get = buildGet(rowKey, classMapping.getAll());
                Result result = table.get(get);
                list.add(ReflectHBaseUtils.parseObject(result, clazz));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return list;
    }

    @Override
    public List<T> scan(TableName tableName) {
        HBaseClassMapping classMapping = ReflectHBaseUtils.getHBaseClassMapping(clazz);
        Scan scan = buildScan(classMapping.getAll());
        return scan(tableName, scan);
    }


    @Override
    public List<T> scan(TableName tableName, Scan scan) {

        List<T> list = new ArrayList<>();
        try (
                Connection connection = instance.getConnection();
                Table table = connection.getTable(tableName)
        ) {
            ResultScanner scanner = table.getScanner(scan);
            for (Result result : scanner) {
                T t = ReflectHBaseUtils.parseObject(result, clazz);
                list.add(t);
            }
            scanner.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return list;
    }

    @Override
    public List<T> scan(TableName tableName, byte[] rowPrefix) {
        HBaseClassMapping classMapping = ReflectHBaseUtils.getHBaseClassMapping(clazz);
        Scan scan = buildScan(classMapping.getAll(), rowPrefix);
        return scan(tableName, scan);
    }

    @Override
    public List<T> scan(TableName tableName, byte[] startRow, byte[] stopRow) {
        HBaseClassMapping classMapping = ReflectHBaseUtils.getHBaseClassMapping(clazz);
        Scan scan = buildScan(classMapping.getAll(), startRow, stopRow);
        return scan(tableName, scan);
    }


    @Override
    public void delete(TableName tableName, byte[] rowKey) {
        try (
                Connection connection = instance.getConnection();
                Table table = connection.getTable(tableName)
        ) {
            Delete delete = new Delete(rowKey);
            table.delete(delete);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void delete(TableName tableName, List<byte[]> rowKeys) {
        try (
                Connection connection = instance.getConnection();
                Table table = connection.getTable(tableName)
        ) {
            List<Delete> deletes = new ArrayList<>();
            for (byte[] rowKey : rowKeys) {
                deletes.add(new Delete(rowKey));
            }
            table.delete(deletes);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void put(TableName tableName, T entity) {

    }

    @Override
    public void put(TableName tableName, List<T> entities) {

    }
}
